package com.baishancloud.log.common.source

import com.baishancloud.log.common.entity._
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.util.ExternalDataFetchUtil
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeHint, TypeInformation}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable

/**
 * 获取外部缓存数据的广播流
 *
 * @author ziqiang.wang 
 * @date 2022/2/18 11:46
 */
object BroadcastUtil extends Serializable {


  /**
   * 拿到自定义要获取的外部资源广播流
   *
   * @param whichInfoFetch 自定义要获取哪些外部资源，将要获取的资源对应的位置变量设置为true即可
   * @param outInterval    广播频率，默认1小时，单位：小时
   */
  def customInfo(whichInfoFetch: ExternalDataEntity, outInterval: Long = 1): BroadcastStream[ExternalDataEntity] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[ExternalDataEntity] {
        var flag: Boolean = true
        val logger: Logger = LoggerFactory.getLogger("broadcast")

        override def run(ctx: SourceFunction.SourceContext[ExternalDataEntity]): Unit = {
          val externalData = ExternalDataEntity()
          while (flag) {
            logger.info("广播流开始获取所有最新外部资源数据")
            if (whichInfoFetch.domainInfoFetch) {
              ExternalDataFetchUtil.domainInfoFetch(externalData.domainInfo)
              logger.info(s"广播流拿到domainInfo数据，map大小为：${externalData.domainInfo.size}")
            }
            if (whichInfoFetch.accountDomainInfoFetch) {
              ExternalDataFetchUtil.accountDomainInfoFetch(whichInfoFetch.account, externalData.accountDomainInfo)
              logger.info(s"广播流拿到accountDomainInfo数据，map大小为：${externalData.accountDomainInfo.size}")
            }
            if (whichInfoFetch.serverCacheGroupInfoFetch) {
              ExternalDataFetchUtil.serverCacheGroupFetch(externalData.serverCacheGroupInfo)
              logger.info(s"广播流拿到服务端cacheGroupInfo数据，map大小为：${externalData.serverCacheGroupInfo.size}")
            }
            if (whichInfoFetch.countryInfoFetch) {
              ExternalDataFetchUtil.countryInfoFetch(externalData.countryInfo)
              logger.info(s"广播流拿到countryInfo数据，map大小为：${externalData.countryInfo.size}")
            }
            if (whichInfoFetch.provinceInfoFetch) {
              ExternalDataFetchUtil.provinceInfoFetch(externalData.provinceInfo)
              logger.info(s"广播流拿到provinceInfo数据，map大小为：${externalData.provinceInfo.size}")
            }
            if (whichInfoFetch.ispInfoFetch) {
              ExternalDataFetchUtil.ispInfoFetch(externalData.ispInfo)
              logger.info(s"广播流拿到ispInfo数据，map大小为：${externalData.ispInfo.size}")
            }
            if (whichInfoFetch.userInfoFetch) {
              ExternalDataFetchUtil.userInfoFetch(externalData.userInfo)
              logger.info(s"广播流拿到userInfo数据，map大小为：${externalData.userInfo.size}")
            }
            if (whichInfoFetch.nodeInfoFetch) {
              ExternalDataFetchUtil.nodeInfoFetch(externalData.nodeInfo)
              logger.info(s"广播流拿到nodeInfo数据，map大小为：${externalData.nodeInfo.size}")
            }
            if (whichInfoFetch.typeModuleInfoFetch) {
              ExternalDataFetchUtil.typeModuleInfoFetch(externalData.typeModuleInfo)
              logger.info(s"广播流拿到typeModuleInfo数据，map大小为：${externalData.typeModuleInfo.size}")
            }
            if (whichInfoFetch.serviceTypeInfoFetch) {
              ExternalDataFetchUtil.serviceTypeInfoFetch(externalData.serviceTypeInfo)
              logger.info(s"广播流拿到serviceTypeInfo数据，map大小为：${externalData.serviceTypeInfo.size}")
            }
            if (whichInfoFetch.svgInfoFetch) {
              ExternalDataFetchUtil.svgInfoFetch(externalData.svgInfo)
              logger.info(s"广播流拿到svgInfo数据，map大小为：${externalData.svgInfo.size}")
            }
            ctx.collect(externalData)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }
      })
      .setParallelism(1)
      .name("customInfo")
      .uid("52a5294a-3d12-4ed4-b6d1-6e0056d2dd03")
      .broadcast(new MapStateDescriptor[String, ExternalDataEntity](
        "customInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[ExternalDataEntity] {})
      ))
  }

  /**
   * 获取所有域名信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def domainInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, DomainEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, DomainEntity]] {

        val logger: Logger = LoggerFactory.getLogger("domainInfo")
        var flag: Boolean = true
        val domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, DomainEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新domainInfo数据")
            ExternalDataFetchUtil.domainInfoFetch(domainInfo)
            logger.info(s"广播流拿到domainInfo数据，map大小为：${domainInfo.size}")
            ctx.collect(domainInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("domainInfo")
      .uid("bfc15751-26b7-408b-8e08-4c5884537f7d")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, DomainEntity]](
        "domainInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, DomainEntity]] {})
      ))
  }

  /**
   * 获取执行账户的域名信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   * @param accounts    账户名称，使用英文逗号分隔
   */
  def accountDomainInfo(outInterval: Long = 1, accounts: String): BroadcastStream[mutable.Map[String, DomainEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, DomainEntity]] {

        val logger: Logger = LoggerFactory.getLogger("accountDomainInfo")
        var flag: Boolean = true
        val accountDomainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, DomainEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新accountDomainInfo数据")
            ExternalDataFetchUtil.accountDomainInfoFetch(accounts, accountDomainInfo)
            logger.info(s"广播流拿到accountDomainInfo数据，map大小为：${accountDomainInfo.size}")
            ctx.collect(accountDomainInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("accountDomainInfo")
      .uid("59b4f633-1a78-4b78-bbc0-fb62d483973e")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, DomainEntity]](
        "accountDomainInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, DomainEntity]] {})
      ))
  }

  /**
   * 获取服务端机器对应cache组信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def serverCacheGroupInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, ServerCacheGroupEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, ServerCacheGroupEntity]] {

        val logger: Logger = LoggerFactory.getLogger("ipCacheGroupInfo")
        var flag: Boolean = true
        val serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, ServerCacheGroupEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新cacheGroupInfo数据")
            ExternalDataFetchUtil.serverCacheGroupFetch(serverCacheGroupInfo)
            logger.info(s"广播流拿到服务端cacheGroupInfo数据，map大小为：${serverCacheGroupInfo.size}")
            ctx.collect(serverCacheGroupInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("serverCacheGroupInfo")
      .uid("d0c4d5b5-6197-4530-ad7b-799d7fe387b7")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, ServerCacheGroupEntity]](
        "serverCacheGroupInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, ServerCacheGroupEntity]] {})
      ))
  }

  /**
   * 获取所有国家信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def countryInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, CountryEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, CountryEntity]] {

        val logger: Logger = LoggerFactory.getLogger("countryInfo")
        var flag: Boolean = true
        val countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, CountryEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新provinceInfo数据")
            ExternalDataFetchUtil.countryInfoFetch(countryInfo)
            logger.info(s"广播流拿到countryInfo数据，map大小为：${countryInfo.size}")
            ctx.collect(countryInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("countryInfo")
      .uid("8bdab765-2a50-49e9-8063-8d3e8c4a5c36")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, CountryEntity]](
        "countryInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, CountryEntity]] {})
      ))
  }


  /**
   * 获取所有省份信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def provinceInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, ProvinceEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, ProvinceEntity]] {

        val logger: Logger = LoggerFactory.getLogger("provinceInfo")
        var flag: Boolean = true
        val provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, ProvinceEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新provinceInfo数据")
            ExternalDataFetchUtil.provinceInfoFetch(provinceInfo)
            logger.info(s"广播流拿到provinceInfo数据，map大小为：${provinceInfo.size}")
            ctx.collect(provinceInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("provinceInfo")
      .uid("ed427245-eb7e-4753-9c0c-0b345a18e4b3")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, ProvinceEntity]](
        "provinceInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, ProvinceEntity]] {})
      ))
  }

  /**
   * 获取所有运营商信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def ispInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, IspEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, IspEntity]] {

        val logger: Logger = LoggerFactory.getLogger("ispInfo")
        var flag: Boolean = true
        val ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, IspEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新ispInfo数据")
            ExternalDataFetchUtil.ispInfoFetch(ispInfo)
            logger.info(s"广播流拿到ispInfo数据，map大小为：${ispInfo.size}")
            ctx.collect(ispInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("ispInfo")
      .uid("a4068e5c-1816-4604-8dc4-116cd5c7c6b7")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, IspEntity]](
        "ispInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, IspEntity]] {})
      ))
  }

  /**
   * 获取所有用户信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认1小时，单位：小时
   */
  def userInfo(outInterval: Long = 1): BroadcastStream[mutable.Map[String, Int]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, Int]] {

        val logger: Logger = LoggerFactory.getLogger("userInfo")
        var flag: Boolean = true
        val userInfo: mutable.Map[String, Int] = mutable.Map[String, Int]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, Int]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新userInfo数据")
            ExternalDataFetchUtil.userInfoFetch(userInfo)
            logger.info(s"广播流拿到userInfo数据，map大小为：${userInfo.size}")
            ctx.collect(userInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("userInfo")
      .uid("286d1c9f-d218-4cff-b09f-a7ed428ddf18")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, Int]](
        "userInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, Int]] {})
      ))
  }

  /**
   * 获取所有主机对应节点和cache组信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认6小时，单位：小时
   */
  def nodeInfo(outInterval: Long = 6): BroadcastStream[mutable.Map[String, NodeEntity]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, NodeEntity]] {

        val logger: Logger = LoggerFactory.getLogger("nodeInfo")
        var flag: Boolean = true
        val nodeInfo: mutable.Map[String, NodeEntity] = mutable.Map[String, NodeEntity]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, NodeEntity]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新nodeInfo数据")
            ExternalDataFetchUtil.nodeInfoFetch(nodeInfo)
            logger.info(s"广播流拿到nodeInfo数据，map大小为：${nodeInfo.size}")
            ctx.collect(nodeInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("nodeInfo")
      .uid("d2ffc2b6-2e17-4015-9dfb-4e2332da6523")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, NodeEntity]](
        "nodeInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, NodeEntity]] {})
      ))
  }

  /**
   * 获取所有主机对应typeModule信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认6小时，单位：小时
   */
  def typeModuleInfo(outInterval: Long = 6): BroadcastStream[mutable.Map[String, mutable.Set[Int]]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, mutable.Set[Int]]] {

        val logger: Logger = LoggerFactory.getLogger("typeModuleInfo")
        var flag: Boolean = true
        val typeModuleInfo: mutable.Map[String, mutable.Set[Int]] = mutable.Map[String, mutable.Set[Int]]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, mutable.Set[Int]]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新typeModuleInfo数据")
            ExternalDataFetchUtil.typeModuleInfoFetch(typeModuleInfo)
            logger.info(s"广播流拿到typeModuleInfo数据，map大小为：${typeModuleInfo.size}")
            ctx.collect(typeModuleInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("typeModuleInfo")
      .uid("f0c81b81-554e-4f16-9644-885a8ce62962")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, mutable.Set[Int]]](
        "typeModuleInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, mutable.Set[Int]]] {})
      ))
  }

  /**
   * 获取所有主机对应服务类型信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认6小时，单位：小时
   */
  def serviceTypeInfo(outInterval: Long = 6): BroadcastStream[mutable.Map[String, Byte]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, Byte]] {

        val logger: Logger = LoggerFactory.getLogger("serviceTypeInfo")
        var flag: Boolean = true
        val serviceTypeInfo: mutable.Map[String, Byte] = mutable.Map[String, Byte]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, Byte]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新serviceTypeInfo数据")
            ExternalDataFetchUtil.serviceTypeInfoFetch(serviceTypeInfo)
            logger.info(s"广播流拿到serviceTypeInfo数据，map大小为：${serviceTypeInfo.size}")
            ctx.collect(serviceTypeInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("serviceTypeInfo")
      .uid("2bc934b2-bb8b-4059-a48a-66fac996f08e")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, Byte]](
        "serviceTypeInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, Byte]] {})
      ))
  }

  /**
   * 获取所有主机对应SVG信息的广播流
   *
   * @param outInterval 广播流数据发射频率，默认6小时，单位：小时
   */
  def svgInfo(outInterval: Long = 6): BroadcastStream[mutable.Map[String, String]] = {
    StreamEnv
      .getStreamExecutionEnvironment
      .addSource(new SourceFunction[mutable.Map[String, String]] {

        val logger: Logger = LoggerFactory.getLogger("svgInfo")
        var flag: Boolean = true
        val svgInfo: mutable.Map[String, String] = mutable.Map[String, String]()

        override def run(ctx: SourceFunction.SourceContext[mutable.Map[String, String]]): Unit = {
          while (flag) {
            logger.info("广播流开始获取最新svgInfo数据")
            ExternalDataFetchUtil.svgInfoFetch(svgInfo)
            logger.info(s"广播流拿到svgInfo数据，map大小为：${svgInfo.size}")
            ctx.collect(svgInfo)
            Thread.sleep(outInterval * 3600 * 1000)
          }
        }

        override def cancel(): Unit = {
          flag = false
        }

      })
      .setParallelism(1)
      .name("svgInfo")
      .uid("b2fc9310-aa71-4951-b4a6-150216983858")
      .broadcast(new MapStateDescriptor[String, mutable.Map[String, String]](
        "svgInfo",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint[mutable.Map[String, String]] {})
      ))
  }

}
